/**
 * 
 */
package cn.com.guage.flink.source;

import java.util.Random;

import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;

import cn.com.guage.flink.domain.SensorReading;

/**
 * @author ASUS
 *
 */
public class SensorSource extends RichParallelSourceFunction<SensorReading> {
    /**
	 * 
	 */
	private static final long serialVersionUID = -475209660791651207L;
	Boolean running = true;
    public void run(SourceContext<SensorReading> sourceContext) throws Exception {
        Random rand = new Random();
        while(running){
            Long curTime = System.currentTimeMillis();
            for (int i = 0; i < 10; i++){
            	Double curTemp = rand.nextGaussian();
                SensorReading sensorReading = new SensorReading("sensor_"+i,curTime,curTemp);
                sourceContext.collect(sensorReading);
            }
            Thread.sleep(3000);
        }
    }
    public void cancel() {
        running = false;
    }
}